package com.caissa.springboot.starter.kafka.consumer;

import com.base.springboot.starter.kafka.consumer.exception.CaissaConsumerDirectStoreErrorMsgException;
import com.base.springboot.starter.kafka.consumer.exception.CaissaConsumerRetryThenStoreErrorMsgException;
import com.base.springboot.starter.kafka.consumer.exception.CaissaIdempotentException;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;

import java.sql.SQLIntegrityConstraintViolationException;

/**
 * 幂等工具类，定义消息重复、业务异常时消息重试、转存的主逻辑
 * User: wangwei
 * Date: 2021-06-15 3:57 下午
 */
@Slf4j
public abstract class CaissaKafkaMsgIdemHelperDepDB extends CaissaKafkaMsgIdemHelper{

    @Autowired
    protected TransactionTemplate transactionTemplate;

    @Override
    protected <R> R doExecute(ConsumerRecord consumerRecord, CBack<R> cBack){
        return transactionTemplate.execute(new TransactionCallback<R>() {
            @Override
            public R doInTransaction(TransactionStatus status) {
                //// 检查幂等
                checkIdempotent(consumerRecord);

                //// 处理业务
                return cBack.doBiz(consumerRecord);
            }
        });
    }

    /**
     * 幂等检查逻辑，基于去重表
     * @param consumerRecord
     * @throws CaissaIdempotentException
     */
    @Override
    protected void checkIdempotent(ConsumerRecord consumerRecord) throws CaissaIdempotentException {
        try {
            insertIdempotent(consumerRecord);
        } catch (Exception e) {

            if (e.getCause() instanceof SQLIntegrityConstraintViolationException) {
                throw new CaissaIdempotentException("", e);
            }

            if (e instanceof CaissaIdempotentException) {
                throw (CaissaIdempotentException)e;
            }

            //其它异常只打日志
            log.error("幂等逻辑执行发生错误，当前没有幂等保护，但不影响主业务逻辑，", e);
        }
    }

    /**
     * 定义去重表写入逻辑
     * @param consumerRecord
     * @throws Exception
     */
    protected abstract boolean insertIdempotent(ConsumerRecord consumerRecord) throws Exception;
}
